package com.ww.exercise.channel.actualcombat.producerconsumer

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong
import kotlin.math.absoluteValue
import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds

class ChannelProcessor<T>(val config: Config = Config()) {
    val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
    val channel = Channel<T>(capacity = config.channelCapacity)

    // 同步计数器管理
    // 生产数
    private val producedCount = AtomicLong(0)

    // 消费数
    val consumedCount = AtomicLong(0)

    // 背压数
    private val backpressureEvents = AtomicLong(0)

    // 关闭表示
    private val isShuttingDown = AtomicBoolean(false)

    // 处理中数
    private val processingCount = AtomicLong(0)

    // 消费者状态管理
    private val consumersStarted = AtomicBoolean(false)

    // 活跃消费者数
    private val activeConsumers = AtomicLong(0)

    /**
     * 计算队列大小 - 考虑正在处理中的项目
     */
    private val calculatedQueueSize: Long
        get() = (producedCount.get() - consumedCount.get()).coerceAtLeast(0)

    /**
     * 启动消费者 - 防止重复启动
     */
    fun startConsumers(processor: suspend (T) -> Unit) {
        if (!consumersStarted.compareAndSet(false, true)) {
            println("⚠️ 消费者已创建，避免重复创建")
            return
        }

        println("🚀 开始创建 ${config.consumerCount} 个消费者...")

        repeat(config.consumerCount) { consumerId ->
            scope.launch {
                activeConsumers.incrementAndGet()
                try {
                    for (item in channel) {
                        processItem(consumerId, item, processor)
                    }
                    println("✅ 消费者${consumerId}正常执行完成")
                } catch (e: CancellationException) {
                    println("🔴 消费者${consumerId}被取消${e.message}")
                } catch (e: Exception) {
                    println("🔴 消费者${consumerId}异常退出${e.message}")
                } finally {
                    activeConsumers.decrementAndGet()
                    println("ℹ️ 消费者${consumerId}关闭")
                }
            }
        }

        println("✅ 所有消费者成功创建")
    }

    /**
     * 处理单个项目 - 支持取消检查和异常处理
     */
    private suspend fun processItem(consumerId: Int, item: T, processor: suspend (T) -> Unit) {
        processingCount.incrementAndGet()
        try {
            // 处理前检查是否已取消
            ensureActive()

            // 记录开始时间用于监控
            val startTime = System.currentTimeMillis()

            // 执行处理
            processor(item)

            // 检查处理是否超时（用于监控）
            val processingTime = System.currentTimeMillis() - startTime
            if (processingTime > config.slowProcessThreshold.inWholeMilliseconds) {
                println("🐌 消费者${consumerId}处理慢，耗时: ${processingTime}ms")
            }

            // 处理完成后再次检查（针对可能的长耗时清理操作）
            ensureActive()
        } catch (e: CancellationException) {
            // 捕获取消异常，优雅退出
            println("⏹️ 消费者${consumerId}处理中被取消: ${e.message}")
            throw e // 重新抛出，确保协程正确结束
        } catch (e: Exception) {
            println("❌ 消费者${consumerId}处理中异常: ${e.message}")
            // 这里可以添加自定义错误处理逻辑
        } finally {
            // 确保计数器递减
            consumedCount.incrementAndGet()
            processingCount.decrementAndGet()
        }
    }

    /**
     * 辅助函数：检查协程是否活跃
     */
    private suspend fun ensureActive() {
        if (!currentCoroutineContext().isActive) {
            throw CancellationException("Coroutine is no longer active")
        }
    }

    /**
     * 生产
     */
    suspend fun produce(item: T): Boolean {
        if (isShuttingDown.get()) {
            println("⏹️ 生产者已关闭，拒绝生产消息：$item")
            return false
        }

        return try {
            channel.send(item)
            // 只在发送成功后计数
            producedCount.incrementAndGet()
            true
        } catch (e: Exception) {
            println("❌ 生产者异常: $item, error: ${e.message}")
            false
        }
    }

    /**
     * 带超时的生产
     */
    suspend fun produceWithTimeout(item: T, timeout: Duration = config.produceTimeout): Boolean {
        return try {
            withTimeout(timeout) {
                produce(item)
            }
        } catch (e: TimeoutCancellationException) {
            backpressureEvents.incrementAndGet()
            println("⏰ 生产者耗时：${timeout}超时: $item")
            false
        }
    }

    /**
     * 非阻塞尝试生产
     */
    fun tryProduce(item: T): Boolean {
        if (isShuttingDown.get()) {
            return false
        }

        val result = channel.trySend(item)
        return if (result.isSuccess) {
            producedCount.incrementAndGet()
            true
        } else {
            if (result.isClosed) {
                println("📬 Channel关闭不支持生产者生产消息: $item")
            }
            false
        }
    }

    /**
     * 批量生产
     */
    suspend fun produceBatch(items: List<T>): BatchResult {
        var successCount = 0
        var failureCount = 0

        for (item in items) {
            if (produceWithTimeout(item)) {
                successCount++
            } else {
                failureCount++
            }
        }

        return BatchResult(successCount, failureCount)
    }

    /**
     * 获取背压状态
     */
    fun getBackpressureStatus(): BackpressureStatus {
        val produced = producedCount.get()
        val consumed = consumedCount.get()
        val processing = processingCount.get()
        val calculatedSize = (produced - consumed).coerceAtLeast(0)

        val isFull = calculatedSize >= config.channelCapacity

        return BackpressureStatus(
            queueSize = calculatedSize.toInt(),
            capacity = config.channelCapacity,
            utilization = calculatedSize.toDouble() / config.channelCapacity,
            isFull = isFull,
            isEmpty = calculatedSize == 0L,
            produced = produced,
            consumed = consumed,
            processing = processing
        )
    }

    /**
     * 获取消费者状态
     */
    fun getConsumerStatus(): ConsumerStatus {
        return ConsumerStatus(
            started = consumersStarted.get(),
            activeCount = activeConsumers.get().toInt(),
            configuredCount = config.consumerCount,
            allActive = activeConsumers.get() == config.consumerCount.toLong()
        )
    }

    /**
     * 优雅停止 - 改进版本：考虑消费者状态
     */
    suspend fun shutdown(timeout: Duration = config.shutdownTimeout): ShutdownResult {
        if (!isShuttingDown.compareAndSet(false, true)) {
            val status = getBackpressureStatus()
            val consumerStatus = getConsumerStatus()
            return ShutdownResult(false, status.queueSize, 0, consumerStatus.activeCount)
        }

        val startTime = System.currentTimeMillis()
        println("🛑 开始优雅关闭...")

        val initialStatus = getBackpressureStatus()
        val initialConsumerStatus = getConsumerStatus()

        println(
            "Initial state: produced=${initialStatus.produced}, consumed=${initialStatus.consumed}, " +
                    "queue=${initialStatus.queueSize}, processing=${initialStatus.processing}, " +
                    "activeConsumers=${initialConsumerStatus.activeCount}"
        )

        // 1. 停止接受新任务
        channel.close()
        println("📬 Channel已关闭，停止接收生产新消息")

        // 2. 等待剩余任务处理完成
        val remainingTasks = waitForCompletion(timeout)

        // 3. 取消所有协程
        scope.cancel()

        val duration = System.currentTimeMillis() - startTime
        val finalStatus = getBackpressureStatus()
        val finalConsumerStatus = getConsumerStatus()

        val result = ShutdownResult(
            success = remainingTasks <= 0 && finalStatus.processing == 0L && finalConsumerStatus.activeCount == 0,
            remainingTasks = remainingTasks,
            shutdownDuration = duration,
            remainingConsumers = finalConsumerStatus.activeCount
        )

        println("📊 关闭结果: $result")
        println(
            "Final state: produced=${finalStatus.produced}, consumed=${finalStatus.consumed}, " +
                    "queue=${finalStatus.queueSize}, processing=${finalStatus.processing}, " +
                    "activeConsumers=${finalConsumerStatus.activeCount}"
        )
        return result
    }

    /**
     * 等待所有任务完成 - 改进版本：考虑消费者状态
     */
    private suspend fun waitForCompletion(timeout: Duration): Int {
        return withTimeoutOrNull(timeout) {
            var lastProgressTime = System.currentTimeMillis()
            var lastQueueSize = calculatedQueueSize
            var lastProcessing = processingCount.get()
            var lastActiveConsumers = activeConsumers.get()
            var consecutiveNoProgress = 0

            while (lastQueueSize > 0 || lastProcessing > 0 || lastActiveConsumers > 0) {
                delay(100)

                val currentSize = calculatedQueueSize
                val currentProcessing = processingCount.get()
                val currentActiveConsumers = activeConsumers.get()

                val hasProgress = currentSize < lastQueueSize ||
                        currentProcessing < lastProcessing ||
                        currentActiveConsumers < lastActiveConsumers

                if (hasProgress) {
                    lastProgressTime = System.currentTimeMillis()
                    lastQueueSize = currentSize
                    lastProcessing = currentProcessing
                    lastActiveConsumers = currentActiveConsumers
                    consecutiveNoProgress = 0

                    if (currentSize % 10 == 0L || currentSize < 10) {
                        println("📈 关闭进度: queue=$currentSize, processing=$currentProcessing, activeConsumers=$currentActiveConsumers")
                    }
                } else {
                    consecutiveNoProgress++
                    if (consecutiveNoProgress > 10) {
                        println("⏰ 1秒无进度: queue=$currentSize, processing=$currentProcessing, activeConsumers=$currentActiveConsumers")
                        break
                    }
                }

                if (currentSize <= 0 && currentProcessing <= 0 && currentActiveConsumers <= 0) {
                    println("✅ 所有消息和消费者已完成关闭")
                    break
                }
            }
            lastQueueSize.toInt()
        } ?: run {
            val remaining = calculatedQueueSize.toInt()
            val processing = processingCount.get()
            val active = activeConsumers.get()
            println("⏰ 关闭超时 $timeout")
            println(" - channel剩余消息数: $remaining")
            println(" - 处理中消息数: $processing")
            println(" - 活跃消费者数: $active")
            println(" - 生产者消息数: ${producedCount.get()}")
            println(" - 消费者消息数: ${consumedCount.get()}")
            remaining
        }
    }

    /**
     * 立即停止
     */
    fun shutdownNow(): ShutdownResult {
        isShuttingDown.set(true)
        scope.cancel()
        channel.close()

        val remaining = calculatedQueueSize.toInt()
        val processing = processingCount.get()
        val active = activeConsumers.get()

        println("🛑 马上关闭")
        println(" - channel剩余消息数: $remaining")
        println(" - Interrupted处理中消息数: $processing")
        println(" - 活跃消费者数: $active")
        println(" - 生产者消息数: ${producedCount.get()}")
        println(" - 消费者消息数: ${consumedCount.get()}")

        return ShutdownResult(
            success = false,
            remainingTasks = remaining,
            shutdownDuration = 0,
            remainingConsumers = active.toInt()
        )
    }

    /**
     * 重启消费者（在某些场景下可能需要）
     */
    fun restartConsumers(processor: suspend (T) -> Unit): Boolean {
        if (!isShuttingDown.get()) {
            println("⚠️ 仍由消费者活跃不允许重启")
            return false
        }

        if (consumersStarted.compareAndSet(true, false)) {
            println("🔄 重启消费者...")
            // 重置状态以便重新启动
            isShuttingDown.set(false)
            startConsumers(processor)
            return true
        }

        return false
    }

    /**
     * 获取监控指标
     */
    fun getMetrics(): Metrics {
        val status = getBackpressureStatus()
        val consumerStatus = getConsumerStatus()
        return Metrics(
            produced = status.produced,
            consumed = status.consumed,
            queueSize = status.queueSize,
            processingCount = status.processing.toInt(),
            backpressureEvents = backpressureEvents.get(),
            isHealthy = !status.isFull && status.queueSize >= 0 && consumerStatus.health == ConsumerHealth.HEALTHY,
            activeConsumers = consumerStatus.activeCount,
            consumerHealth = consumerStatus.health
        )
    }

    /**
     * 获取详细状态（用于调试）
     */
    fun getDetailedStatus(): DetailedStatus {
        val status = getBackpressureStatus()
        val consumerStatus = getConsumerStatus()
        return DetailedStatus(
            produced = status.produced,
            consumed = status.consumed,
            calculatedQueueSize = status.queueSize.toLong(),
            processingCount = status.processing,
            isShuttingDown = isShuttingDown.get(),
            channelClosed = channel.isClosedForSend,
            consumersStarted = consumerStatus.started,
            activeConsumers = consumerStatus.activeCount,
            consumerHealth = consumerStatus.health
        )
    }

    /**
     * 检查处理器是否活跃
     */
    fun isActive(): Boolean {
        return !isShuttingDown.get() && scope.isActive && consumersStarted.get()
    }

    /**
     * 重置处理器（用于测试）
     */
    @OptIn(ExperimentalStdlibApi::class)
    fun resetForTesting() {
        if (!isShuttingDown.get()) {
            shutdownNow()
        }

        // 重置所有状态
        producedCount.set(0)
        consumedCount.set(0)
        backpressureEvents.set(0)
        processingCount.set(0)
        activeConsumers.set(0)
        isShuttingDown.set(false)
        consumersStarted.set(false)
    }
}

/**
 * 配置类
 */
data class Config(
    val consumerCount: Int = Runtime.getRuntime().availableProcessors(),
    val channelCapacity: Int = 1000,
    val produceTimeout: Duration = 5000.milliseconds,
    val slowProcessThreshold: Duration = 1000.milliseconds,
    val shutdownTimeout: Duration = 30.seconds,
    val cancellationCheckInterval: Duration = 100.milliseconds
)

/**
 * 运行指标
 */
data class Metrics(
    val produced: Long,
    val consumed: Long,
    val queueSize: Int,
    val processingCount: Int,
    val backpressureEvents: Long,
    val isHealthy: Boolean,
    val activeConsumers: Int = 0,
    val consumerHealth: ConsumerHealth = ConsumerHealth.NOT_STARTED
) {
    val totalLag: Long get() = produced - consumed
    val activeItems: Int get() = queueSize + processingCount
    val successRate: Double get() = if (produced > 0) consumed.toDouble() / produced else 1.0
    val throughput: Double get() = if (produced > 0) consumed.toDouble() / produced else 0.0

    override fun toString(): String {
        return "Metrics(produced=$produced, consumed=$consumed, queue=$queueSize, " +
                "processing=$processingCount, backpressure=$backpressureEvents, " +
                "healthy=$isHealthy, consumers=$activeConsumers, consumerHealth=$consumerHealth, " +
                "lag=$totalLag, active=$activeItems, successRate=${"%.2f".format(successRate * 100)}%)"
    }
}

data class BackpressureStatus(
    val queueSize: Int,
    val capacity: Int,
    val utilization: Double,
    val isFull: Boolean,
    val isEmpty: Boolean,
    val produced: Long,
    val consumed: Long,
    val processing: Long
) {
    val hasBackpressure: Boolean get() = isFull
    val severity: BackpressureSeverity
        get() = when {
            utilization >= 0.9 -> BackpressureSeverity.HIGH
            utilization >= 0.75 -> BackpressureSeverity.MEDIUM
            utilization >= 0.5 -> BackpressureSeverity.LOW
            else -> BackpressureSeverity.NONE
        }
    val countersConsistent: Boolean get() = (produced - consumed) >= 0

    override fun toString(): String {
        return "BackpressureStatus(queue=$queueSize/$capacity (${"%.1f".format(utilization * 100)}%), " +
                "full=$isFull, empty=$isEmpty, produced=$produced, consumed=$consumed, " +
                "processing=$processing, severity=$severity)"
    }
}

enum class BackpressureSeverity {
    NONE,
    LOW,
    MEDIUM,
    HIGH;

    fun shouldAlert(): Boolean = this == MEDIUM || this == HIGH
}

/**
 * 消费者状态信息
 */
data class ConsumerStatus(
    val started: Boolean,
    val activeCount: Int,
    val configuredCount: Int,
    val allActive: Boolean
) {
    val health: ConsumerHealth
        get() = when {
            !started -> ConsumerHealth.NOT_STARTED
            activeCount == 0 -> ConsumerHealth.NO_ACTIVE
            activeCount < configuredCount -> ConsumerHealth.DEGRADED
            else -> ConsumerHealth.HEALTHY
        }

    override fun toString(): String {
        return "ConsumerStatus(started=$started, active=$activeCount/$configuredCount, health=$health)"
    }
}

enum class ConsumerHealth {
    NOT_STARTED,    // 消费者未启动
    NO_ACTIVE,      // 没有活跃消费者
    DEGRADED,       // 部分消费者异常
    HEALTHY         // 所有消费者正常
}

data class BatchResult(
    val successCount: Int,
    val failureCount: Int
) {
    val total: Int get() = successCount + failureCount
    val successRate: Double get() = if (total > 0) successCount.toDouble() / total else 0.0

    override fun toString(): String {
        return "BatchResult(success=$successCount, failed=$failureCount, " +
                "successRate=${"%.1f".format(successRate * 100)}%)"
    }
}

data class ShutdownResult(
    val success: Boolean,
    val remainingTasks: Int,
    val shutdownDuration: Long,
    val remainingConsumers: Int = 0
) {
    val message: String
        get() = if (success) {
            "✅ Shutdown completed successfully in ${shutdownDuration}ms"
        } else {
            "⚠️ Shutdown incomplete: $remainingTasks tasks and $remainingConsumers consumers remaining after ${shutdownDuration}ms"
        }

    override fun toString(): String {
        return "ShutdownResult(success=$success, remainingTasks=$remainingTasks, " +
                "remainingConsumers=$remainingConsumers, duration=${shutdownDuration}ms)"
    }
}

/**
 * 详细状态（用于调试）
 */
data class DetailedStatus(
    val produced: Long,
    val consumed: Long,
    val calculatedQueueSize: Long,
    val processingCount: Long,
    val isShuttingDown: Boolean,
    val channelClosed: Boolean,
    val consumersStarted: Boolean,
    val activeConsumers: Int,
    val consumerHealth: ConsumerHealth
) {
    val totalUnprocessed: Long get() = (produced - consumed).coerceAtLeast(0)
    val activeItems: Long get() = calculatedQueueSize + processingCount
    val countersConsistent: Boolean get() = (produced - consumed) >= 0
    val consistencyDiff: Long get() = (produced - consumed - calculatedQueueSize).absoluteValue

    override fun toString(): String {
        return "DetailedStatus(produced=$produced, consumed=$consumed, " +
                "queue=$calculatedQueueSize, processing=$processingCount, " +
                "shuttingDown=$isShuttingDown, channelClosed=$channelClosed, " +
                "consumersStarted=$consumersStarted, activeConsumers=$activeConsumers, " +
                "consumerHealth=$consumerHealth, unprocessed=$totalUnprocessed, " +
                "active=$activeItems, consistent=$countersConsistent, diff=$consistencyDiff)"
    }
}

/**
 * 使用示例和测试代码
 */
object ChannelProcessorDemo {
    @JvmStatic
    fun main(args: Array<String>) = runBlocking {
        val processor = ChannelProcessor<Int>(
            Config(
                consumerCount = 4,
                channelCapacity = 100,
                produceTimeout = 2.seconds,
                shutdownTimeout = 10.seconds
            )
        )

        // 启动消费者
        processor.startConsumers { item ->
            // 模拟处理时间
            delay((100..500).random().toLong())
            println("Processed: $item on ${Thread.currentThread().name}")
        }

        // 尝试重复启动（应该被阻止）
        processor.startConsumers { item ->
            println("This should not be called: $item")
        }

        // 生产数据
        val producer = launch {
            for (i in 1..50) {
                val success = processor.produceWithTimeout(i)
                if (!success) {
                    println("Failed to produce: $i")
                }
                delay(50) // 控制生产速度
            }
            println("Producer finished")
        }

        // 监控
        val monitor = launch {
            while (producer.isActive) {
                delay(1000)
                val metrics = processor.getMetrics()
                val status = processor.getBackpressureStatus()
                val consumerStatus = processor.getConsumerStatus()
                println("=== 监控 ===")
                println("Metrics: $metrics")
                println("背压状态: $status")
                println("消费者状态: $consumerStatus")
                println("===============")
            }
        }

        producer.join()
        monitor.cancel()

        // 优雅关闭
        val shutdownResult = processor.shutdown()
        println(shutdownResult.message)
    }
}

/**
 * 高级使用示例：带错误处理和恢复
 */
object AdvancedChannelProcessorDemo {
    @JvmStatic
    fun main(args: Array<String>) = runBlocking {
        val processor = ChannelProcessor<String>(
            Config(
                consumerCount = 2,
                channelCapacity = 50
            )
        )

        var errorCount = 0

        // 启动消费者，模拟偶尔的错误
        processor.startConsumers { item ->
            try {
                // 模拟10%的失败率
                if (Math.random() < 0.1) {
                    errorCount++
                    throw RuntimeException("Simulated processing error for: $item")
                }

                // 模拟处理时间
                delay((200..800).random().toLong())
                println("✅ 成功消费了: $item")

            } catch (e: Exception) {
                println("❌ 消费中错误： $item: ${e.message}")
                // 在实际应用中，这里可以添加重试逻辑或错误报告
            }
        }

        // 批量生产数据
        val producer = launch {
            val items = (1..100).map { "Item-$it" }
            val result = processor.produceBatch(items)
            println("批量生产结果: $result")
        }

        // 高级监控
        val monitor = launch {
            var lastProduced = 0L
            while (producer.isActive) {
                delay(2000)
                val metrics = processor.getMetrics()
                val detailedStatus = processor.getDetailedStatus()

                val currentThroughput = metrics.consumed - lastProduced
                lastProduced = metrics.consumed

                println("=== Advanced Monitor ===")
                println("吞吐量 (last 2s): $currentThroughput items")
                println("总共错误次数: $errorCount")
                println("Detailed: $detailedStatus")
                println("=======================")
            }
        }

        producer.join()
        monitor.cancel()

        // 显示最终状态
        println("=== 最终状态 ===")
        println("Metrics: ${processor.getMetrics()}")
        println("Consumer Status: ${processor.getConsumerStatus()}")

        // 优雅关闭
        val shutdownResult = processor.shutdown()
        println(shutdownResult.message)
    }
}